package spark.util;

import org.apache.hadoop.hive.ql.io.orc.OrcSerde;
import org.apache.hadoop.hive.ql.io.orc.OrcStruct;
import org.apache.hadoop.hive.serde2.objectinspector.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
import org.apache.hadoop.io.Writable;
import scala.Serializable;
import spark.OrcReadAndWrite.Kryo.ORCFormat;

import java.util.ArrayList;
import java.util.List;

/**
 * 用于在Mapper中读ORC文件
 * @author Administrator
 */
public class ORCUtilBak implements Serializable{
//public class ORCUtil {
//	ORC结构体数据类型
	private OrcStruct orc = null;

//	OrcSerde.serialize()的参数
	private StructObjectInspector structObjectInspector = null;

//	OrcSerde.serialize()的参数
	public List<Object> list = null;

//	写出的数据
	private OrcSerde orcSerde = null;

//	private TypeInfo typeInfo = TypeInfoUtils.getTypeInfoFromTypeString(ORCFormat.INS_STATUS);
//	private ObjectInspector oi = TypeInfoUtils.getStandardJavaObjectInspectorFromTypeInfo(typeInfo);
//	structObjectInspector = (StructObjectInspector)oi;

	public void setRecord(OrcStruct orc){
		this.orc = orc;
	}
	/**
	 * @param type String类型的Struct描述
	 *             得到每个字段
	 */
	public void setOrcType(String type){
//		typeInfo = TypeInfoUtils.getTypeInfoFromTypeString(type);
//		structObjectInspector = (StructObjectInspector)OrcStruct.createObjectInspector(typeInfo);
		TypeInfo typeInfo = TypeInfoUtils.getTypeInfoFromTypeString(ORCFormat.INS_STATUS);
		this.structObjectInspector = (StructObjectInspector)TypeInfoUtils.getStandardJavaObjectInspectorFromTypeInfo(typeInfo);
	}
	/**
	 * 根据ORC一行记录获得某个字段的值
	 * @param key hive表中的某个字段
	 * @return 字段对应的具体value
	 */
	public String getDate(String key){
		StructField structFieldRef = structObjectInspector.getStructFieldRef(key);
		Object data = structObjectInspector.getStructFieldData(this.orc, structFieldRef);
		String value = String.valueOf(data);
		if (value == null || value.toLowerCase() == "null") {
			return null;
		}
		return value;
	}

	/**
	 * 写数据调用的方法，返回orc类型的数据写出
	 */
	public Writable serialize(){
		if (orcSerde == null) {
			orcSerde = new OrcSerde();
		}
		Writable line = orcSerde.serialize(list, structObjectInspector);
		list = new ArrayList<Object>();
		//System.out.println("序列化");
		return line;
	}

	/**
	 * 向List中添加字段
	 */
	public ORCUtilBak addField(Object field){
		if (list == null ) {
			list = new ArrayList<Object>();
		}
			list.add(field);
		return this;
	}
}
